﻿using cn.sunchis.pvuv.sdk.model;
using Confluent.Kafka;
using Confluent.Kafka.Serialization;
using Newtonsoft.Json;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Net.Http;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using System.Web.Configuration;

namespace cn.sunchis.pvuv.sdk.config
{
    /// <summary>
    /// PVUV Kafka生产者
    /// <para>@Author 许亮</para>
    /// <para>@Create 2018-7-4 16:55:20</para>
    /// </summary>
    internal class PvuvKafkaProducer
    {
        /// <summary>
        /// Kafka队列主题
        /// </summary>
        private const string TOPIC = "CCPG-LOG-PLATFORM-PVUV";
        /// <summary>
        /// Kafka连接配置
        /// </summary>
        private Dictionary<string, object> kafkaConfig;
        /// <summary>
        /// Kafka生产者对象
        /// </summary>
        private Producer<string, string> kafkaProducer;

        public PvuvKafkaProducer() 
        {
            this.kafkaConfig = new Dictionary<string, object>() {
                {"bootstrap.servers", WebConfigurationManager.AppSettings["KafkaServers"]}
            };
            this.kafkaProducer = new Producer<string, string>(this.kafkaConfig, new StringSerializer(Encoding.UTF8), new StringSerializer(Encoding.UTF8));
        }

        public void newVisit(HttpRequestMessage request, CancellationToken cancellationToken) 
        {
            Pvuv pvuv = new Pvuv();
            pvuv.sys = "C# .NET-ASP应用";
            pvuv.com = "pvuv-demo-asp-donet";
            pvuv.setHost(Environment.MachineName, request.RequestUri.Port);
            pvuv.url = request.RequestUri.PathAndQuery;
            pvuv.stamp = DateTime.Now;

            var delivery = this.kafkaProducer.ProduceAsync(TOPIC, "pvuv", JsonConvert.SerializeObject(pvuv));
            delivery.ContinueWith(task => { 

            });
        }
    }
}
